package com.distributer.delayqueue.delayqueue.ready;

import java.util.List;
import java.util.TimerTask;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.distributer.delayqueue.delayqueue.ConsumeQueueProvider;
import com.distributer.delayqueue.delayqueue.JobOperationService;
import com.distributer.delayqueue.delayqueue.Queue;
import com.distributer.delayqueue.delayqueue.event.EventBus;
import com.distributer.delayqueue.delayqueue.event.RedisJobTraceEvent;
import com.distributer.delayqueue.delayqueue.support.DistributedLock;
import com.distributer.delayqueue.delayqueue.support.RedisQueueProperties;
import com.distributer.delayqueue.job.JobMsg;
import com.distributer.delayqueue.job.JobStatus;
import com.distributer.delayqueue.util.NamedUtil;

/**
 * 实时队列任务,到达时间后将任务发送到mq中
 */
public class ReadyTimeTask extends TimerTask{

	private static final Logger LOGGER = LoggerFactory.getLogger(ReadyTimeTask.class);
	private RedisQueueProperties redisQueueProperties;
	private JobOperationService jobOperationService;
	private Queue delayQueue;
	private DistributedLock lock;
	private ConsumeQueueProvider consumeQueueProvider;

	@Override
	public void run() {
		runTemplate();
	}
	
	private void runTemplate(){
		if (redisQueueProperties.isCluster()) {
			String lockName = NamedUtil.buildLockName(NamedUtil.buildRealTimeName(redisQueueProperties.getProfix(), redisQueueProperties
                    .getName(), redisQueueProperties.getReadyName()));
			try{
				lock.lock(lockName);
				runInstance();
			}finally{
				lock.unlock(lockName);
			}
		}else{
			runInstance();
		}
	}
	
	
	private void runInstance(){
		List<String> jobIds = jobOperationService.getReadyJob(10);
		if (jobIds != null && jobIds.size()>0) {
			for (String jobId : jobIds) {
				JobMsg job = jobOperationService.getJob(jobId);
				if (job == null) {
					LOGGER.info("该job:{}没有数据!",jobId);
					jobOperationService.removeReadyJob(jobId);
					continue;
				}
				try {
					if (lock.lock(job.getJobId())) {
						//任务已删除
						if (job.getStatus() == JobStatus.Delete.ordinal()) {
							LOGGER.info("该job:{}已经被删除!",jobId);
							jobOperationService.removeJobToPool(jobId);
							continue;
						}
						if (!check(job)) {
							//未到执行时间重新加入到延时队列
							delayQueue.putJobMsg(job);
							continue;
						}
						//消费队列
						consumeQueueProvider.comsumer(job);
						job.setStatus(JobStatus.Finish.ordinal());
						jobOperationService.updateJobStatus(jobId, JobStatus.Finish);
						jobOperationService.removeReadyJob(jobId);
						jobOperationService.removeJobToPool(jobId);
						EventBus.getInstance().postEvent(new RedisJobTraceEvent(job));
					}
				}catch(Exception e){
					LOGGER.error("执行实时任务失败！",e);
				} finally{
					lock.unlock(job.getJobId());
				}
			}
		}
	}

	private boolean check(JobMsg job) {
		long executeTime = job.getDelayTime() + job.getCreateTime().getTime(),
				currentTime = System.currentTimeMillis();
		if (executeTime < currentTime) {
			return true;
		}
		return false;
	}

	public void setRedisQueueProperties(RedisQueueProperties redisQueueProperties) {
		this.redisQueueProperties = redisQueueProperties;
	}

	public void setJobOperationService(JobOperationService jobOperationService) {
		this.jobOperationService = jobOperationService;
	}

	public void setDelayQueue(Queue delayQueue) {
		this.delayQueue = delayQueue;
	}

	public void setLock(DistributedLock lock) {
		this.lock = lock;
	}

	public void setConsumeQueueProvider(ConsumeQueueProvider consumeQueueProvider) {
		this.consumeQueueProvider = consumeQueueProvider;
	}

}
